package storm;


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import storm.TwitterSpout;

public class Main {

	/**
	 * @param args
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws InterruptedException {

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("spout", new TwitterSpout(), 1);
		
		builder.setBolt("count",new TokenizerBolt(),1).allGrouping("spout");
		builder.setBolt("stop", new StopBolt(),1).allGrouping("count");
		builder.setBolt("sugg", new SuggesterBolt(),1).allGrouping("stop");
		builder.setBolt("stem", new StemmerBolt(),1).allGrouping("sugg");
		
		Config conf = new Config();

		conf.setMaxTaskParallelism(6);

		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("cooccurrence", conf, builder.createTopology());

	}

}
